package net.bwie.realtime.jtp.dwd.log.function1;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.utils.DateTimeUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

public class AdjustIsNewProcessFunction extends KeyedProcessFunction<String,String,String> {
    //声明状态
    private transient ValueState<String> firstDateValue;
    @Override
    public void open(Configuration parameters) throws Exception {
        //初始化状态值
        firstDateValue = getRuntimeContext().getState(
                new ValueStateDescriptor<String>("first-date", String.class)
        );
    }

    @Override
    public void processElement(String value,Context ctx, Collector<String> out) throws Exception {
        //获取状态值
        String firstDate = firstDateValue.value();
        //解析字段
        JSONObject jsonObject = JSON.parseObject(value);
        String isNew = jsonObject.getJSONObject("common").getString("is_new");
        Long ts = jsonObject.getLong("ts");
        //为新用户时
        if("1".equals(isNew)){
            //状态值为null时
            if (firstDate == null){
                //赋值
                firstDateValue.update(DateTimeUtil.convertLongToString(ts,DateTimeUtil.DATE_FORMAT));
            }else {
                //时间相同不变,不同修改值
                if(!firstDate.equals(DateTimeUtil.convertLongToString(ts,DateTimeUtil.DATE_FORMAT))){
                    jsonObject.getJSONObject("common").put("is_new","0");
                }
            }
        }else {
            if(isNew == null){
                firstDateValue.update(DateTimeUtil.convertLongToString(ts - 86400000,DateTimeUtil.DATE_FORMAT));
            }
        }
        out.collect(jsonObject.toJSONString());
    }
}
